package com.dd.stack.kafka.producer.service.inner.impl;


import com.dd.stack.kafka.common.pojo.BO.JsonUserBO;
import com.dd.stack.kafka.common.pojo.BO.LogMessage;
import com.dd.stack.kafka.common.pojo.BO.User;
import com.dd.stack.kafka.producer.util.KafkaProducerUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZoneOffset;

/**
 * @Author liuxianmeng
 * @CreateTime 2025/2/14 2:39
 * @Description KafkaServiceImpl
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class KafkaServiceImpl {

    private final KafkaTemplate<String, Object> kafkaTemplateObj;

    /********************************** Simple Test *********************************/

    // 定义 Kafka 主题
    private static final String TOPIC_TEST = "test_topic";

    /**
     * 发送消息到 Kafka 此处是生产者
     * @param message 要发送的消息内容
     */
    public void sendMessage(String message) {
        kafkaTemplateObj.send(TOPIC_TEST, message);
    }

    /******************************** avro序列化Java对象 ********************************/

    private static final String TOPIC_USER = "users";

    //public void sendUser(User user) {
    //    // 转换为Avro生成的类
    //    com.dd.stack.kafka.common.pojo.avro.gene.User avroUserAvro = com.dd.stack.kafka.common.pojo.avro.gene.User.newBuilder()
    //            .setId(user.getId())
    //            .setUsername(user.getUsername())
    //            .setEmail(user.getEmail())
    //            .setCreatedAt(
    //                    Instant.ofEpochSecond(user.getCreatedAt()
    //                    .atZone(ZoneId.systemDefault()).toInstant().toEpochMilli())
    //             )
    //            .build();
    //    kafkaTemplateObj.send(TOPIC_USER, avroUserAvro);
    //}

    /******************************** Json序列化Java对象 ********************************/

    public void sendJsonUser(JsonUserBO userBO) {
        KafkaProducerUtil.sendMessage(TOPIC_USER, userBO);
    }

    /******************************** Kafka处理日志 ********************************/

    private static final String LOG_TOPIC = "app-logs";

    public void sendLog(LogMessage logMessage) {
        KafkaProducerUtil.sendMessage(LOG_TOPIC, logMessage);
    }
}
